package com.zongxingh.falcon.mq.producer.impl;

import com.zongxingh.falcon.mq.config.MqProperties;
import com.zongxingh.falcon.mq.producer.MqProducer;
import com.zongxingh.falcon.mq.utils.ExceptionTools;
import com.zongxingh.falcon.mq.utils.MqUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

/**
 * Introduction:  <简介>
 * Package Name:  com.zongxingh.falcon.mq.producer.impl
 * Project Name:  mq
 * Author:  zongxingh
 * Email： zongxingh@163.com
 * Tel: 157*****778
 * Create Time:  2020/9/22 15:14
 */
public class RocketMqProducer extends MqProducer {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqProducer.class);

    private DefaultMQProducer producer;

    public RocketMqProducer() {
    }

    public RocketMqProducer(String producerGroup, MqProperties properties) {
        this.producerGroup = producerGroup;
        this.name = producerGroup;
        // 实例化一个生产者来产生延时消息
        MqProperties.Rocketmq rocketmq = properties.getRocketmq();
        this.producer = new DefaultMQProducer(this.producerGroup);
        this.producer.setNamesrvAddr(rocketmq.getNamesrvAddr());
        this.producer.setClientCallbackExecutorThreads(rocketmq.getClientCallbackExecutorThreads());
        this.producer.setRetryTimesWhenSendFailed(rocketmq.getProducer().getRetryTimesWhenSendFailed());
        // 启动生产者
        try {
            producer.start();
        } catch (MQClientException e) {
            LOGGER.error("启动RocketMq的producer失败，失败原因：{}", ExceptionTools.getStackTraceAsString(e));
        }
    }

    public DefaultMQProducer getProducer() {
        return producer;
    }

    public void setProducer(DefaultMQProducer producer) {
        this.producer = producer;
    }

    @Override
    public boolean sedMessage(String topic, String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        LOGGER.debug("向队列发送消息，topic为：{}，消息内容为：{}", topic, text);
        Message message = new Message(topic, text.getBytes());
        SendResult result = this.producer.send(message);
        return getSendResult(result);
    }

    @Override
    public boolean sedMessage(String topic, byte[] bytes) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        LOGGER.debug("向队列发送消息，topic为：{}，消息内容为：{}", topic, new String(bytes, StandardCharsets.UTF_8));
        Message message = new Message(topic, bytes);
        SendResult result = this.producer.send(message);
        return getSendResult(result);
    }

    private boolean getSendResult(SendResult result) {
        if (SendStatus.SEND_OK.equals(result.getSendStatus())) {
            return true;
        } else if (SendStatus.FLUSH_DISK_TIMEOUT.equals(result.getSendStatus())) {
            LOGGER.error("发送消息失败，消息同步到磁盘超时");
            return false;
        } else if (SendStatus.FLUSH_SLAVE_TIMEOUT.equals(result.getSendStatus())) {
            LOGGER.error("发送消息失败，消息同步到从节点失败");
            return false;
        } else if (SendStatus.SLAVE_NOT_AVAILABLE.equals(result.getSendStatus())) {
            LOGGER.error("发送消息失败，消息同步到从节点，从节点不可获得");
            return false;
        } else {
            LOGGER.error("发送消息失败，发生未知错误");
            return false;
        }
    }

}
